from lxml import etree
from queue import Queue
from multiprocessing import Pool, Manager
from requests.auth import HTTPProxyAuth
from tool import *
from sql import sqlseverDB
import sys,traceback
import math,time,logging


logger = logging.getLogger()
logger.setLevel(logging.INFO)  # 设置日志级别
log_path=f'日志文件/亚马逊CJASIN.log'
file_handler = logging.FileHandler(log_path, mode='a', encoding='utf-8')

# 创建控制台处理器 (Handler)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)

# 设置日志格式
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)

# 将处理器添加到 logger 对象
logger.addHandler(file_handler)
logger.addHandler(console_handler)
ms = sqlseverDB('{SQL Server}', SQL_SERVER, 'SuMaiTongPol', SQL_USER, SQL_PWD)
mutil_count=5
def process_shop(bid,kitems,queue,task_count):
    logging.info(f"进程启动: bid={bid}, 任务数={len(kitems)}")
    try:
        res= asyncio.run(cjAsin(bid,kitems,queue,task_count))
        logging.info(res)

    except Exception as e:
        logging.error(f'bid:{bid}, 出现错误: {e}')
        traceback.print_exc()
        time.sleep(3)



async def cjAsin(bid,kitems,ycj_queue,task_count):

    browser=await openSunBrowser(bid)
    if not browser:
        return
    page=await browser.newPage()
    for kitem in kitems:
        suc = 0
        fail = 0
        kid, kw = kitem
        x_ind=ycj_queue.qsize()+1
        print(f'第{x_ind}个,关键词:《{kw}》,开始采集...')
        asinList = []
        for p in range(1, 8):
            if p == 1:
                url = f'https://www.amazon.com/s?k={kw}&s=date-desc-rank&ref=sr_st_date-desc-rank'
            else:
                url = f'https://www.amazon.com/s?k={kw}&s=date-desc-rank&page={p}&qid={int(time.time())}&ref=sr_pg_{p}'
            try:
                await page.goto(url,{'timeout':0})
                await asyncio.sleep(2)
                is_has_result=False
                wait_count=0
                while True:
                    cur_tit= await page.title()
                    if cur_tit=='Sorry! Something went wrong!':
                        logging.info(f'第{x_ind}个,kid:{kid},kw:《{kw}》,页面503')
                        break
                    cot= await page.content()
                    if len(cot)<20000:
                        logging.info(f'第{x_ind}个,kid:{kid},kw:《{kw}》,页面内容过少')
                        break

                    
                    el1=await page.xpath('//span[@class="a-color-state a-text-bold"]')
                    el2=await page.xpath('//span[text()="No results for "]')
                    el3=await page.xpath('//div[@data-asin!="" and @data-index]/@data-asin')
                    if el1 or el3:
                        is_has_result=True
                        break
                    if el2:
                        logging.info(f'第{x_ind}个,kid:{kid},kw:《{kw}》,无搜索结果退出采集')
                        break
                    if wait_count>30:
                        logging.info(f'第{x_ind}个,kid:{kid},kw:《{kw}》,等待过久退出')
                        break
                    wait_count+=1
                    await asyncio.sleep(1)
                if not is_has_result:
                    break

            except Exception as e:
                print(f'第{x_ind}个,{kw},第{p}页,出错 => {e}')
                traceback.print_exc()
            cur_asins = etree.HTML(cot).xpath('//div[@data-asin!="" and @data-index]/@data-asin')
            asinList.extend(cur_asins)
            await asyncio.sleep(1)

        asinList = set(asinList)

        print(f'第{x_ind}个,关键词:《{kw}》,共7页,采集了 {len(asinList)} 个url')
        for asin in asinList:
            purl = f'https://www.amazon.com/dp/{asin}?th=1&psc=1'
            rs = ms.ExecQuery('select count(*) from AmazonAsin where asin=?', (asin,))
            if rs[0][0] == 0:
                aff = ms.ExecNoQuery('insert into AmazonAsin([ASIN],[Purl],[SearchKw]) values(?,?,?)', (asin, purl, kw))
                if aff > 0:
                    suc += 1
                else:
                    fail += 1
            else:
                fail += 1
        aff_cg=0
        if len(asinList)>0:
            aff_cg=ms.ExecNoQuery('update AmazonKW set state=1,isuse=isuse+1 where id=?',(kid,))
        ycj_queue.put(1)
        latest_count=task_count-ycj_queue.qsize()
        print(f'第{x_ind}个,kid:{kid},共《{kw}》,共保存:{suc}个链接,更改采集状态:{aff_cg},剩余任务队列:{latest_count}/{task_count} \n')

    await page.close()
def main():
 
    print(f'第一步采集关键词....')
    while True:
        rs_isuse=ms.ExecQuery('select min(isuse) from AmazonKW') #先获取采集最小次数
        min_use_count=rs_isuse[0][0]
        rs = ms.ExecQuery(f'select id, kw from amazonkw where state=0 and isuse={min_use_count}')
        if len(rs)>0:
            break
        else:
            aff=ms.ExecNoQuery('update amazonkw set state=0')
            print(f'关键词最小采集次数:{min_use_count},已经全部采集完毕,将所有 {aff} 个关键词采集状态置0')

    taskCount = len(rs)

    print(f'本次采集,关键词最小使用次数:{min_use_count},待采集关键词总数: {taskCount}')

    #创建采集环境
    while True:
        all_bids=get_browser_environment()
        print('获取到',all_bids)
        if len(all_bids)>=mutil_count:
            bids=all_bids[:mutil_count]
            break
        else:
            print(mutil_count-len(all_bids))
            for _ in range(mutil_count-len(all_bids)):
                res=create_browser_environment_no_proxy(f'亚马逊采集{get_random_code()}')
                time.sleep(1)
    
    print(f'成功创建:{len(bids)}个采集环境')

    manager = Manager()
    queue = manager.Queue()

    qu_jian=math.ceil(taskCount/mutil_count)
    with Pool(processes=mutil_count) as pool:  # 根据实际情况设置并发进程数量
        results = []
        for j in range(mutil_count):
            kitems=rs[j*qu_jian:(j+1)*qu_jian]
            result = pool.apply_async(
                process_shop, args=(bids[j],kitems,queue,taskCount))
            results.append(result)

        # 监控任务执行情况
        pool.close()
        for result in results:
            result.wait()  # 等待每个进程完成，无超时限制

        # 所有任务执行完成后继续
        pool.join()



if __name__ == '__main__':
    main()
